package com.sh.data.engine.domain.shims.kafka.util;

import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.sh.data.engine.common.util.RedisKeyUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @author: mengzheng.mhc
 * @date: 2024/6/11 14:48
 */
@Slf4j
public class KafkaUtil {

    private static final String ARG_ERROR_MSG = "kafka server配置有误，请检查";

    public static void testConnection(String address, Integer timeout)
        throws IOException, IllegalArgumentException {
        if (timeout == null) {
            timeout = 10;
        }

        try {
            List<Pair<String, Integer>> list = SocketUtil.parseHostAndPort(address, 2181);
            SocketUtil.testSocketConnection(list, timeout);
        } catch (IllegalArgumentException e) {
            throw new IllegalArgumentException(ARG_ERROR_MSG, e);
        }
    }

    public static Set<String> kafkaTopic(String dsLink) {
        Set<String> resultSet = Sets.newHashSet();
        if (StringUtils.isBlank(dsLink)) {
            return resultSet;
        }
        AdminClient client = null;

        RedisTemplate redisTemplate = SpringUtil.getBean("redisTemplate", RedisTemplate.class);


        try {
            String cacheKey = RedisKeyUtil.getKafkaTopicCacheKeyByDsLink(dsLink);
            Object topicsJsonObj = redisTemplate.opsForValue().get(cacheKey);
            if (topicsJsonObj instanceof String) {
                String topicsJsonStr = (String) topicsJsonObj;
                if (StringUtils.isNotBlank(topicsJsonStr)) {
                    List<String> topics = JSONUtil.toList(topicsJsonStr, String.class);
                    resultSet.addAll(topics);
                    return resultSet;
                }
            }

            Properties pro = new Properties();
            pro.put("bootstrap.servers", dsLink);
            pro.put("connections.max.idle.ms", 10000);
            pro.put("request.timeout.ms", 8000);
            client = KafkaAdminClient.create(pro);
            ListTopicsResult result = client.listTopics();
            KafkaFuture<Set<String>> set = result.names();
            resultSet = set.get();
            if (resultSet == null) {
                resultSet = Sets.newHashSet();
            }
            redisTemplate.opsForValue().set(cacheKey, JSONUtil.toJsonStr(resultSet), 1, TimeUnit.MINUTES);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
        return resultSet;
    }

    public static List<String> preview(String hosts, String topic) {
        // 初始化消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", hosts);
        props.put("group.id", UUID.randomUUID().toString().replaceAll("-", ""));
        //props.put("group.id", "user_testa");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "10000");
        props.put("max.poll.records", "1");
        props.put("session.timeout.ms", "120000");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));

        //消费1秒数据
        //ConsumerRecords<String, String> records = consumer.poll(1000);

        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
        ArrayList<String> data = Lists.newArrayList();
        records.forEach(record -> {
            data.add(record.value());
        });
        return data;
    }

    public static List<String> preview(
        String hosts, List<String> topics, Integer maxRecordCount, long timeout) {
        Properties props = new Properties();

        props.put("bootstrap.servers", hosts);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", "preview_data_group");

        props.put("enable.auto.commit", "true");
        props.put("max.poll.records", 10);
        props.put("auto.commit.interval.ms", 1000);
        props.put("auto.offset.reset", "earliest");

        List<String> results = new ArrayList<>();

        try {
            KafkaUtil.testConnection(hosts, 3);
        } catch (IOException e) {
            log.warn("kafka数据源不在线");
            return results;
        }

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {

            consumer.subscribe(topics);

            long totalTime = 0;
            long pollTimeout = 100;
            while (true) {
                if (results.size() >= maxRecordCount || totalTime >= timeout) {
                    break;
                }
                long s = System.currentTimeMillis();
                ConsumerRecords<String, String> records = consumer.poll(pollTimeout);
                long time = System.currentTimeMillis() - s;
                for (ConsumerRecord<String, String> record : records) {
                    if (results.size() >= maxRecordCount) {
                        break;
                    }
                    results.add(record.value());
                }

                totalTime += time;
            }
        }

        if (results.size() > maxRecordCount) {
            results = Lists.partition(results, maxRecordCount).get(0);
        }

        return results;
    }

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "47.104.197.234:9092");
        props.put("group.id", "hr_test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("max.poll.records", "1");
        props.put("session.timeout.ms", "120000");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//
        String topic = "wmc00";
        int msgCount = 1000;

        org.apache.kafka.clients.producer.KafkaProducer<String, String> kafkaProducer =
            new org.apache.kafka.clients.producer.KafkaProducer<>(props);

        for (int i = 1; i < msgCount; i++) {
            //      OpcData data = Util.getData();
            //
            //      String msg = JSON.toJSONString(data);
            String msg = "{\"id\":%d,\"name\":\"呵呵哈哈哈\"}";
            int index = i;
            String format = String.format(msg, i);
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, format);

            kafkaProducer.send(
                record,
                new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        System.out.println("========第" + index + "条消息发布成功=======:" + format);
                    }
                });
        }
        kafkaProducer.close();
        // 初始化消费者
//    Properties props2 = new Properties();
//    props2.put("bootstrap.servers", "47.104.197.234:9092");
//    //props2.put("group.id", UUID.randomUUID().toString().replaceAll("-", ""));
//    //28faf94f647a4b97
//    props2.put("group.id", "28faasdjhafdll");
//    props2.put("enable.auto.commit", "true");
//    props2.put("auto.commit.interval.ms", "10000");
//    props2.put("max.poll.records", "1");
//    props2.put("session.timeout.ms", "120000");
//    props2.put("auto.offset.reset", "earliest");
//    props2.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//    props2.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props2);
//    consumer.subscribe(Arrays.asList(topic));
//    String targetTopic="wmc0613";
//    //消费5秒数据
//    while (true){
//      ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
//
//      ArrayList<String> data = Lists.newArrayList();
//      records.forEach(record->{
//        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(targetTopic,record.value());
////        kafkaProducer.send(
////                producerRecord,
////                new Callback() {
////                  @Override
////                  public void onCompletion(RecordMetadata metadata, Exception exception) {
////                    //System.out.println("========第" + index + "条消息发布成功=======:"+format);
////                  }
////                });
//        data.add(record.value());
//      });
////      if(data.size()>5){
////        break;
////      }
//      System.out.println(data);
//    }

    }
}
